package com.atguigu.tingshu.album.mq;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.service.TrackInfoService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class TrackReceiver {

    @Autowired
    private TrackInfoService trackInfoService ;

    @Autowired
    private RedisTemplate<String , String> redisTemplate ;

    // 幂等性
    // 通过redis 临时存储唯一id,如果保存失败 则证明之前接受过消息,是重复消息读取,则保存失败,反之则成功
    @KafkaListener(topics = KafkaConstant.QUEUE_TRACK_STAT_UPDATE)
    public void trackStatNumListener(ConsumerRecord<String , String> consumerRecord) {

        String record = consumerRecord.value();
        if(!StringUtils.isEmpty(record)) {

            TrackStatMqVo trackStatMqVo = JSON.parseObject(record, TrackStatMqVo.class);
            String businessNo = trackStatMqVo.getBusinessNo();

            // setIfAbsent方法表示的意思就是如何key是否存在的设置失败，如果key是不存在的就设置成功，底层使用的redis中的setnx命令
            Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent("track:play:" + businessNo, "1", 1, TimeUnit.HOURS);
            if(Boolean.TRUE.equals(ifAbsent)) {
                // 如果是true就说明成功了，是第一次消费
                trackInfoService.updateStatNum(trackStatMqVo) ;
            }

        }

    }

}